/**
 * Copyright (c) Huawei Technologies Co., Ltd. 2022. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * Description: Define api of stream cache.
 */
#ifndef DATASYSTEM_CLIENT_STREAM_CACHE_CLIENT_WORKER_API_H
#define DATASYSTEM_CLIENT_STREAM_CACHE_CLIENT_WORKER_API_H

#include <memory>
#include <string>

#include "datasystem/common/flags/flags.h"
#include "datasystem/client/client_worker_common_api.h"
#include "datasystem/client/mmap_manager.h"
#include "datasystem/client/stream_cache/receive_element.h"
#include "datasystem/common/shared_memory/shm_unit.h"
#include "datasystem/common/shared_memory/shm_unit_info.h"
#include "datasystem/common/stream_cache/stream_data_page.h"
#include "datasystem/common/util/net_util.h"
#include "datasystem/common/util/status_helper.h"
#include "datasystem/protos/worker_stream.pb.h"
#include "datasystem/protos/stream_posix.stub.rpc.pb.h"
#include "datasystem/stream/consumer.h"
#include "datasystem/stream/element.h"
#include "datasystem/utils/optional.h"
#include "datasystem/utils/sensitive_value.h"

namespace datasystem {
namespace client {
namespace stream_cache {
class ClientWorkerApi : public ClientWorkerCommonApi {
public:
    /**
     * @brief Construct ClientWorkerApi.
     * @param[in] hostPort The address of the worker node.
     * @param[in] cred The authentication credentials.
     * @param[in] signature Used to do AK/SK authenticate.
     * @param[in] tenantId TenantId of client user.
     *
     */
    ClientWorkerApi(const HostPort &hostPort, RpcCredential cred,
                    Signature *signature = nullptr, std::string tenantId = "");

    /**
     * @brief Initialize the ClientWorkerApi Object.
     * @param[in] requestTimeoutMs Request milliseconds.
     * @param[in] connectTimeoutMs Connect timeout milliseconds.
     * @return K_OK on success; the error code otherwise.
     *         K_INVALID: the input ip or port is invalid.
     */
    Status Init(int32_t timeoutMs, int32_t connectTimeoutMs);

    /**
     * @brief Send rpc request to worker to create one producer.
     * @param[in] streamName The name of stream.
     * @param[in] producerId The producer id generate by client.
     * @param[in] producerConf The stream configurations, including pageSize, maxStreamSize, etc.
     * @param[out] senderProducerNo The producer number generated by master.
     * @param[out] enableStreamDataVerification Should data verification be on.
     * @param[out] outPageView ShmView of cursor.
     * @param[out] outStreamMetaView ShmView of streamMetaShm.
     * @return Status of the call.
     */
    Status CreateProducer(const std::string &streamName, const std::string &producerId,
                          const ProducerConf &producerConf, ShmView &outPageView,
                          DataVerificationHeader::SenderProducerNo &senderProducerNo,
                          bool &enableStreamDataVerification, uint64_t &streamNo, bool &enableSharedPage,
                          uint64_t &sharedPageSize, ShmView &outStreamMetaView);

    /**
     * @brief Send rpc request to worker to create one consumer.
     * @param[in] streamName The name of stream.
     * @param[in] consumerId The consumer id generate by client.
     * @param[in] config The configure of subscription, such as subscription name, subscription mode.
     * @param[out] rsp The response of the subscription
     * @return Status of the call.
     */
    Status Subscribe(const std::string &streamName, const std::string &consumerId,
                     const struct SubscriptionConfig &config, SubscribeRspPb &rsp);

    /**
     * @brief Send rpc request to worker to delete stream.
     * @param[in] streamName The name of stream that will be delete.
     * @return Status of the call.
     */
    Status DeleteStream(const std::string &streamName);

    /**
     * @brief Query global producer count by streamName.
     * @param[in] streamName Target stream.
     * @param[out] producerNum The producer count.
     * @return Status of the call.
     */
    Status QueryGlobalProducersNum(const std::string &streamName, uint64_t &producerNum);

    /**
     * @brief Query global consumer count by streamName.
     * @param[in] streamName Target stream.
     * @param[out] consumerNum The consumer count.
     * @return Status of the call.
     */
    Status QueryGlobalConsumersNum(const std::string &streamName, uint64_t &consumerNum);

    /**
     * @brief Send rpc request to worker to reset the provided streams.
     * @param[in] streamNames Target streams.
     * @return Status of the call.
     */
    Status ResetStreams(const std::vector<std::string> &streamNames);

    /**
     * @brief Send rpc request to worker to resume the provided streams.
     * @param[in] streamNames Target streams.
     * @return Status of the call.
     */
    Status ResumeStreams(const std::vector<std::string> &streamNames);

    /**
     * @brief Construct log prefix.
     * @return Return the log prefix.
     */
    std::string LogPrefix() const;

    /**
     * @brief Ensure that the latest client ID is obtained.
     * @return Return the client id.
     */
    std::string GetClientId();

private:
    friend class ProducerConsumerWorkerApi;
    /**
     * @brief To check and reset the timeout of receive.
     * @param[in] requestedTimeout The client specified timeout value for time spent in the system.
     * @param[out] rpcTimeout The timeout used for RPC.
     * @param[out] adjustedTimeout The timeout used in worker.
     * @return Status of the call.
     */
    Status SetRpcTimeout(int64_t &requestedTimeout, int32_t &rpcTimeout, int64_t &adjustedTimeout);

    std::unique_ptr<ClientWorkerSCService_Stub> rpcSession_;
};
}  // namespace stream_cache
}  // namespace client
}  // namespace datasystem
#endif
